#ifndef __M_CHANNEL_H__
#define __M_CHANNEL_H__
#include"../mqcommon/mq_helper.hpp"
#include"../mqcommon/mq_logger.hpp"
#include"../mqcommon/mq_msg.pb.h"
#include"../mqcommon/mq_proto.pb.h"
#include"../mqcommon/mq_threadpool.hpp"

#include"mq_consumer.hpp"
#include"mq_exchage.hpp"
#include"mq_virtualhost.hpp"

 #include "muduo/proto/codec.h" 
#include "muduo/proto/dispatcher.h"
#include"muduo/net/TcpConnection.h"
#include"mq_route.hpp"
namespace bitmq{

    using ProtobufCodecPtr=std::shared_ptr<ProtobufCodec>;

    using openChannelRequestPtr=std::shared_ptr<openChannelRequest>;
    using closeChannelRequestPtr=std::shared_ptr<closeChannelRequest>;
    using declareExchangeRequestPtr=std::shared_ptr<declareExchangeRequest>;
    using deleteExchangeRequestPtr=std::shared_ptr<deleteExchangeRequest>;
    using declareQueueRequestPtr=std::shared_ptr<declareQueueRequest>;
    using deleteQueueRequestPtr=std::shared_ptr<deleteQueueRequest>;
    using queueBindingRequestPtr=std::shared_ptr<queueBindingRequest>;
    using queueUnBindingRequestPtr=std::shared_ptr<queueUnBindingRequest>;
    using basicPublishRequestPtr=std::shared_ptr<basicPublishRequest>;
    using basicAckRequestPtr=std::shared_ptr<basicAckRequest>;
    using basicConsumerRequestPtr=std::shared_ptr<basicConsumerRequest>;
    using basicConcelRequestPtr=std::shared_ptr<basicConcelRequest>;
    class Channel{
        public:
        using ptr=std::shared_ptr<Channel>;
        Channel(const std::string &cid,
            const muduo::net::TcpConnectionPtr&connection,
            const ProtobufCodecPtr&codec,
            const VirtualHost::ptr&vhost,
            const threadpool::ptr&pool,
            const ConsumerManager::ptr& cmp)
            :_cid(cid)
            ,_connection(connection)
            ,_codec(codec)
            ,_vhost(vhost)
            ,_pool(pool)
            ,_cmp(cmp){}
        ~Channel()
        {
            if(_consumer.get()!=nullptr)
            {
                _cmp->remove(_consumer->_tag,_consumer->_qname);
            }
        }
         //定义和删除交换机
        void declareExchange(const declareExchangeRequestPtr&req)
        {
            // printf("exchange_type: %d\n", req->type());
            bool ret=_vhost->declareExchange(req->exchange_name(),req->type(),req->durable(),req->auto_delete(),req->args());
            return basicResponse(ret,req->rid(),req->cid());
        }
        void deleteExchange(const deleteExchangeRequestPtr&req)
        {
            _vhost->deleteExchange(req->exchange_name());
            return basicResponse(true,req->rid(),req->cid());
        }

        //定义和删除队列
        void declareQueue(const declareQueueRequestPtr&req)
        {
            bool ret=_vhost->declareQueue(req->queue_name(),req->durable(),req->exclusive(),req->auto_delete(),req->args());
            if (ret == false) {
                return basicResponse(false, req->rid(), req->cid());
            }
            _cmp->initQueueConsumer(req->queue_name());//初始化消费者队列
            return basicResponse(true,req->rid(),req->cid());
        }
        void deleteQueue(const deleteQueueRequestPtr&req)
        {
            _cmp->destroyQueueConsumer(req->queue_name());

            _vhost->deleteQueue(req->queue_name());
            return basicResponse(true,req->rid(),req->cid());
        }

        //绑定和解绑
        void bind(const queueBindingRequestPtr&req)
        {
            bool ret=_vhost->bind(req->exchange_name(),req->queue_name(),req->binding_key());
            return basicResponse(ret,req->rid(),req->cid());
        }
        void unbind(const queueUnBindingRequestPtr&req)
        {
            _vhost->unbind(req->exchange_name(),req->queue_name());
            return basicResponse(true,req->rid(),req->cid());
        }
        
        //发布消息和消费消息
        void basicPublish(const basicPublishRequestPtr&req)//消息是发布给交换机的，再由交换机路由分配给队列
        {
            //1、取出一台交换机
            Exchage::ptr exchange=_vhost->selectExchange(req->exchange_name());
            if( exchange.get()==nullptr )
            {

                return basicResponse(false,req->rid(),req->cid());
            }
            //2、再取出交换机绑定的所有队列
            // printf("exchange_name: %s\n", req->exchange_name().c_str());
            QueueBingdingMap qbm=_vhost->getbinds(req->exchange_name());
            BasicProperties* properties=nullptr;
            std::string routing_key;
            bool ret = false;
            ret=req->has_properties();
            if(ret)
            {
                properties=req->mutable_properties();
                routing_key=properties->routing_key();
                //std::cout<<"r_k1:"<<routing_key<<" ";
            }
            //std::cout<<req-
            for(auto &binding:qbm)
            {
                //2.1路由分配
                if(bitmq::Router::route(exchange->_type,routing_key,binding.second->_bingding_key))
                {
                    //2.2将消息添加到队列中（消息管理）
                
                    std::cout<<"type:"<<exchange->_type<<"r_k2:"<<routing_key<<" "<<"b_k2:"<<binding.second->_bingding_key<<std::endl;
                 
                    _vhost->basicPublic(binding.first,properties,req->body());
                    //2.3将一个消息处理任务交给线程池（向指定的订阅者推送消息---线程池完成）
                    auto task=std::bind(&Channel::consumer,this,binding.first);
                    _pool->push(task);
                }
            
            }
            return basicResponse(true, req->rid(), req->cid());
        }
        //消息确认
        void basicAck(const basicAckRequestPtr&req)
        {
            _vhost->basicAck(req->queue_name(),req->msg_id());
            return basicResponse(true,req->rid(),req->cid());
        }
        //订阅队列消息
        void basicConsumer(const basicConsumerRequestPtr&req)
        {
            //判断队列是否存在
            bool ret=_vhost->existsQueue(req->queue_name());
            if(ret==false)
            {
                return basicResponse(false,req->rid(),req->cid());
            }
            //创建消费者添加到消费者管理中
            auto task=std::bind(&Channel::callback,this,std::placeholders::_1,std::placeholders::_2,std::placeholders::_3);
            _consumer=_cmp->creater(req->consumer_tag(),req->queue_name(),req->auto_ack(),task);

            return basicResponse(true,req->rid(),req->cid());
        }
        
        //取消订阅
       
        void basicConcel(const basicConcelRequestPtr&req)
        {
            _cmp->remove(req->consumer_tag(),req->queue_name());
            return basicResponse(true,req->rid(),req->cid());
        }
        private:
        void callback(const std::string&tag,const bitmq::BasicProperties*bp,const std::string&body)
        {
            basicConsumerRespose resp;
            resp.set_consumer_tag(tag);
            resp.set_body(body);
            resp.set_cid(_cid);
            if(bp)
            {
                resp.mutable_properties()->set_id(bp->id());
                resp.mutable_properties()->set_delivery_mode(bp->delivery_mode());
                resp.mutable_properties()->set_routing_key(bp->routing_key());
            }
            _codec->send(_connection,resp);
        }
        void consumer(const std::string&qname)
        {
            //1、从队列里取出一条消息
            MessagePtr msg=_vhost->basicConsume(qname);
            if(msg.get()==nullptr)
            {
                DLOG("没找到消费的消息");
                return;
            }
            //2、选出一名消费者对象
            Consumer::ptr cp=_cmp->choose(qname);
            if(cp.get()==nullptr)
            {
                DLOG("没找到消费者");
                return;
            }
            //3、调用消费者处理消息---实现消息的推送
            cp->_callback(cp->_tag,msg->mutable_payload()->mutable_properties(),msg->payload().body());
            //4、判断是否自动删除---是，就无需等待直接删除   否，等待消费者确认删除
            if(cp->_auto_ack) _vhost->basicAck(qname,msg->payload().properties().id());
        }
        void basicResponse(bool ok,const std::string&rid,const std::string&cid)
        {
            basicCommonRespose rep;
            rep.set_ok(ok);
            rep.set_rid(rid);
            rep.set_cid(cid);
             _codec->send(_connection,rep);
        }
        private:
        std::string _cid;//信道id
        Consumer::ptr _consumer;//信道相连的消费者
        muduo::net::TcpConnectionPtr _connection;//连接管理句柄
        ProtobufCodecPtr _codec;//协议处理句柄
        VirtualHost::ptr _vhost;//虚拟机管理句柄
        threadpool::ptr _pool;
        ConsumerManager::ptr _cmp;//消费者管理句柄
    };
    class ChannelManager{
        public:
        using ptr=std::shared_ptr<ChannelManager>;
        ChannelManager(){}
        bool openChannel(const std::string &cid,
            const muduo::net::TcpConnectionPtr&connection,
            const ProtobufCodecPtr&codec,
            const VirtualHost::ptr&vhost,
            const threadpool::ptr&pool,
            const ConsumerManager::ptr& cmp){
                std::unique_lock<std::mutex> lock(_mutex);
                auto it=_channels.find(cid);
                if(it!=_channels.end())
                {
                    DLOG("信道%s,没找到!",cid.c_str());
                    return false;
                }

                auto channel=std::make_shared<Channel>(cid,connection,codec,vhost,pool,cmp);
                _channels.insert(std::make_pair(cid,channel));
                return true;
            }
        void closeChannel(const std::string&id)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _channels.erase(id);
        }
        Channel::ptr getChannel(const std::string&id)
        {
            std::unique_lock<std::mutex> lock(_mutex);
                auto it=_channels.find(id);
                if(it==_channels.end())
                {
                    DLOG("信道%s,没找到!",id.c_str());
                    return Channel::ptr();
                }
                return it->second;
        }
        private:
        std::mutex _mutex;
        std::unordered_map<std::string,Channel::ptr> _channels;//信道id& 信道
    };
}

#endif